# -*- coding: utf-8 -*-
"""
CPS 数据质量监控 Provider。
"""
from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple

from app.blueprints.reports.providers.base import BaseProvider, SafeDict
from app.services.report_service import ReportTemplateService
from app.utils.logger import FlinkLogger

# 延迟初始化 logger，避免应用上下文问题
logger = None


def get_logger():
    global logger
    if logger is None:
        logger = FlinkLogger.get_logger(__name__)
    return logger


class CPSDataQualityProvider(BaseProvider):
    """CPS基础数据同步专用报表数据提供者"""

    def _get_platform_configs(self) -> Dict[str, Any]:
        """从数据库 default_params 读取平台配置"""
        if not self.template_code:
            raise ValueError("模板编码不能为空")
        
        try:
            tpl = ReportTemplateService.get_template_by_code(self.template_code)
            if not tpl:
                raise ValueError(f"模板 {self.template_code} 不存在")
            
            default_params = tpl.get("default_params")
            if isinstance(default_params, str):
                try:
                    default_params = json.loads(default_params)
                except Exception as e:
                    get_logger().error(f"解析 default_params 失败: {e}")
                    raise ValueError(f"解析模板配置失败: {e}")
            
            if not isinstance(default_params, dict):
                raise ValueError("default_params 必须是字典类型")
            
            platform_configs = default_params.get("platform_configs", {})
            if not platform_configs:
                raise ValueError("模板配置中缺少 platform_configs")
            
            return platform_configs
        except Exception as e:
            get_logger().error(f"读取平台配置失败: {e}", exc_info=True)
            raise

    def _get_sql_template(self) -> str:
        """从数据库 default_params 读取 SQL 模板"""
        if not self.template_code:
            raise ValueError("模板编码不能为空")
        
        try:
            tpl = ReportTemplateService.get_template_by_code(self.template_code)
            if not tpl:
                raise ValueError(f"模板 {self.template_code} 不存在")
            
            default_params = tpl.get("default_params")
            if isinstance(default_params, str):
                try:
                    default_params = json.loads(default_params)
                except Exception as e:
                    get_logger().error(f"解析 default_params 失败: {e}")
                    raise ValueError(f"解析模板配置失败: {e}")
            
            if not isinstance(default_params, dict):
                raise ValueError("default_params 必须是字典类型")
            
            sql_template = default_params.get("sql_template", "")
            if not sql_template:
                raise ValueError("模板配置中缺少 sql_template")
            
            return sql_template
        except Exception as e:
            get_logger().error(f"读取 SQL 模板失败: {e}", exc_info=True)
            raise

    def _get_field_labels(self) -> Dict[str, str]:
        """
        获取字段英文名到中文标题的映射：默认值 -> 模板配置 -> 运行时自定义
        """
        labels: Dict[str, str] = {}
        
        # 默认字段标签
        default_labels = {
            "platform": "平台",
            "platform_summary": "平台汇总",
            "source_layer": "原始层(源表)",
            "source_total_count": "原始层商品总数",
            "dws_total_count": "目标表商品总数",
            "dws_on_sale_count": "目标表上架商品数",
            "dws_off_sale_count": "目标表下架商品数",
            "sync_rate": "同步率",
            "data_status": "数据状态",
        }
        
        # 从模板配置读取
        if self.template_code:
            try:
                tpl = ReportTemplateService.get_template_by_code(self.template_code)
                if tpl:
                    default_params = tpl.get("default_params")
                    if isinstance(default_params, str):
                        try:
                            default_params = json.loads(default_params)
                        except Exception:
                            default_params = {}
                    if isinstance(default_params, dict):
                        template_labels = default_params.get("field_labels", {}) or {}
                        if template_labels:
                            labels = template_labels
            except Exception:
                get_logger().debug(
                    "[CPS数据质量监控] 读取模板字段标签失败", exc_info=True
                )

        # 运行时自定义标签（优先级最高）
        runtime_labels = {}
        if isinstance(self.params, dict):
            runtime_labels = self.params.get("field_labels") or {}

        # 合并：默认 -> 模板 -> 运行时
        merged = default_labels.copy()
        if isinstance(labels, dict):
            merged.update(labels)
        if isinstance(runtime_labels, dict):
            merged.update(runtime_labels)

        get_logger().debug(
            "[CPS数据质量监控] 字段标签合并结果: %s",
            json.dumps(merged, ensure_ascii=False),
        )
        return merged

    def _build_time_condition(self, time_filter: Dict[str, Any]) -> str:
        """构建时间条件 SQL"""
        time_range_type = time_filter.get("time_range_type", "all")
        time_start = time_filter.get("time_start")
        time_end = time_filter.get("time_end")

        time_condition = ""
        if time_range_type == "recent":
            # 最近7天：总是动态计算，确保定时任务自动更新
            time_condition = " AND etl_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)"
        elif time_range_type == "yesterday":
            # 昨天：总是动态计算，确保定时任务自动更新
            now = datetime.now()
            yesterday_start = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
            yesterday_end = (now - timedelta(days=1)).replace(hour=23, minute=59, second=59, microsecond=999999)
            time_condition = f" AND etl_date BETWEEN '{yesterday_start.strftime('%Y-%m-%d %H:%M:%S')}' AND '{yesterday_end.strftime('%Y-%m-%d %H:%M:%S')}'"
        elif time_range_type == "last_month":
            # 上月：总是动态计算，确保定时任务自动更新
            now = datetime.now()
            # 上个月的第一天
            first_day_of_last_month = (now.replace(day=1) - timedelta(days=1)).replace(day=1)
            # 上个月的最后一天
            last_day_of_last_month = now.replace(day=1) - timedelta(days=1)
            last_month_start = first_day_of_last_month.replace(hour=0, minute=0, second=0, microsecond=0)
            last_month_end = last_day_of_last_month.replace(hour=23, minute=59, second=59, microsecond=999999)
            time_condition = f" AND etl_date BETWEEN '{last_month_start.strftime('%Y-%m-%d %H:%M:%S')}' AND '{last_month_end.strftime('%Y-%m-%d %H:%M:%S')}'"
        elif time_range_type == "custom" and time_start and time_end:
            # 自定义：使用用户指定的时间范围
            time_condition = (
                f" AND etl_date BETWEEN '{time_start}' AND '{time_end}'"
            )
        
        return time_condition

    def _build_source_union_sql(
        self, 
        selected_platforms: List[str], 
        platform_configs: Dict[str, Any],
        time_condition: str
    ) -> str:
        """根据平台配置构建源表 UNION SQL"""
        source_union_parts = []
        
        for platform in selected_platforms:
            if platform not in platform_configs:
                get_logger().warning(
                    "[CPS数据质量监控] 跳过不支持的平台: %s", platform
                )
                continue

            platform_config = platform_configs[platform]
            platform_name = platform_config.get("name", platform)
            tables = platform_config.get("tables", [])
            supplement_table = platform_config.get("supplement_table")

            platform_source_tables = []
            for tbl_config in tables:
                table_name = tbl_config.get("table", "")
                cast_expr = tbl_config.get("cast_expr", "product_id")
                use_distinct = tbl_config.get("use_distinct", False)
                
                if not table_name:
                    continue
                
                if use_distinct:
                    platform_source_tables.append(
                        f"SELECT DISTINCT {cast_expr} AS product_id FROM {table_name} WHERE 1=1{time_condition}"
                    )
                else:
                    platform_source_tables.append(
                        f"SELECT {cast_expr} AS product_id FROM {table_name} WHERE 1=1{time_condition}"
                    )

            # 添加补充表
            if supplement_table:
                platform_source_tables.append(
                    f"SELECT product_id FROM {supplement_table} "
                    f"WHERE platform = '{platform}'"
                )

            if platform_source_tables:
                source_sql = f"""
    -- ========== {platform_name}平台({platform}) ==========
    SELECT '{platform}' AS platform,
           COUNT(DISTINCT product_id) AS source_total_count
    FROM (
        {' UNION ALL '.join(platform_source_tables)}
    ) t"""
                source_union_parts.append(source_sql)

        return "\n\nUNION ALL\n\n".join(source_union_parts)

    def _render_sql_template(
        self, 
        sql_template: str, 
        context: Dict[str, Any]
    ) -> str:
        """渲染 SQL 模板"""
        try:
            # 使用 SafeDict 确保缺失的占位符返回空字符串
            safe_context = SafeDict(context)
            rendered_sql = sql_template.format_map(safe_context)
            return rendered_sql
        except Exception as e:
            get_logger().error(f"渲染 SQL 模板失败: {e}", exc_info=True)
            raise ValueError(f"渲染 SQL 模板失败: {e}")

    def collect(self) -> Dict[str, Any]:
        try:
            get_logger().info(
                "[CPS数据质量监控] 开始收集数据"
            )

            data_source_config = self.params.get("data_source_config", {})
            data_source_id = data_source_config.get("data_source_id")

            if not data_source_id:
                raise ValueError("数据源ID不能为空")

            platform_selection = self.params.get("platform_selection") or self.params.get(
                "platform_filter", {}
            )
            selected_platforms = platform_selection.get("selected_platforms", ["dy"])
            include_summary = platform_selection.get("include_summary", True)

            get_logger().debug(
                "[CPS数据质量监控] 平台筛选: selected_platforms=%s, include_summary=%s",
                selected_platforms,
                include_summary,
            )

            time_filter = self.params.get("time_filter", {})
            threshold_config = self.params.get("threshold_config", {})
            sync_rate_threshold = threshold_config.get("sync_rate_threshold", 80)

            # 构建时间条件
            time_condition = self._build_time_condition(time_filter)

            # 从数据库读取平台配置
            platform_configs = self._get_platform_configs()

            # 构建源表 UNION SQL（按用户选择的具体 platform 统计）
            source_union_sql = self._build_source_union_sql(
                selected_platforms, platform_configs, time_condition
            )

            if not source_union_sql:
                get_logger().warning(
                    "[CPS数据质量监控] 没有平台数据，selected_platforms=%s",
                    selected_platforms,
                )
                final_sql = (
                    "SELECT '无数据' AS '平台', '无平台选择' AS '平台汇总', "
                    "'无数据' AS '原始层(源表)', 0 AS '原始层商品总数', "
                    "0 AS '目标表商品总数', 0 AS '目标表上架商品数', 0 AS '目标表下架商品数', "
                    "'0%' AS '同步率', '无数据' AS '数据状态', 1 AS sort_order"
                )
            else:
                # 获取字段标签
                field_labels = self._get_field_labels()

                # 构建平台汇总 SQL
                platform_summary_sql = ""
                if include_summary:
                    platform_summary_sql = f"""
SELECT
    'ALL' AS `{field_labels.get('platform', '平台')}`,
    '全平台汇总' AS `{field_labels.get('platform_summary', '平台汇总')}`,
    '所有平台汇总' AS `{field_labels.get('source_layer', '原始层(源表)')}`,
    SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) AS `{field_labels.get('source_total_count', '原始层商品总数')}`,
    SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) AS `{field_labels.get('dws_total_count', '目标表商品总数')}`,
    SUM(`{field_labels.get('dws_on_sale_count', '目标表上架商品数')}`) AS `{field_labels.get('dws_on_sale_count', '目标表上架商品数')}`,
    SUM(`{field_labels.get('dws_off_sale_count', '目标表下架商品数')}`) AS `{field_labels.get('dws_off_sale_count', '目标表下架商品数')}`,
    CONCAT(
        ROUND(
            CASE
                WHEN SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) > 0
                THEN (SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) * 100.0 / SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`))
                ELSE 0
            END,
            2
        ),
        '%'
    ) AS `{field_labels.get('sync_rate', '同步率')}`,
    CASE
        WHEN SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) = 0 THEN '无原始数据'
        WHEN SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) = 0 THEN '未同步'
        WHEN (SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) * 100.0 / SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`)) < {sync_rate_threshold} THEN '同步率过低'
        ELSE '正常'
    END AS `{field_labels.get('data_status', '数据状态')}`,
    0 AS sort_order
FROM platform_detail

UNION ALL

"""

                # 获取 SQL 模板
                sql_template = self._get_sql_template()

                # 构建模板上下文
                # DWS 表查询使用用户选择的具体 platform，platform_all 仅用于展示
                formatted_platforms = ", ".join([f"'{p}'" for p in selected_platforms])
                context = {
                    "source_union_sql": source_union_sql,
                    "selected_platforms": formatted_platforms,  # DWS 表使用具体 platform
                    "time_condition": time_condition,
                    "sync_rate_threshold": sync_rate_threshold,
                    "platform_summary_sql": platform_summary_sql,
                    "platform_label": field_labels.get("platform", "平台"),
                    "platform_summary_label": field_labels.get("platform_summary", "平台汇总"),
                    "source_layer_label": field_labels.get("source_layer", "原始层(源表)"),
                    "source_total_label": field_labels.get("source_total_count", "原始层商品总数"),
                    "dws_total_label": field_labels.get("dws_total_count", "目标表商品总数"),
                    "dws_on_sale_label": field_labels.get("dws_on_sale_count", "目标表上架商品数"),
                    "dws_off_sale_label": field_labels.get("dws_off_sale_count", "目标表下架商品数"),
                    "sync_rate_label": field_labels.get("sync_rate", "同步率"),
                    "data_status_label": field_labels.get("data_status", "数据状态"),
                }

                # 渲染 SQL 模板
                final_sql = self._render_sql_template(sql_template, context)

            get_logger().info("执行CPS数据质量监控查询，平台: %s", selected_platforms)

            with self.get_data_source_connection(data_source_id) as conn:
                cursor = conn.cursor()
                cursor.execute(final_sql)
                columns = [desc[0] for desc in cursor.description]
                rows = cursor.fetchall()

                get_logger().info("查询完成，返回 %s 条记录", len(rows))

                rows_list: List[List[Any]] = []
                for row in rows:
                    if isinstance(row, (list, tuple)):
                        rows_list.append(list(row))
                    elif isinstance(row, dict):
                        rows_list.append([row.get(col) for col in columns])
                    else:
                        try:
                            rows_list.append(list(row))
                        except TypeError:
                            rows_list.append([row])

            filtered_columns, filtered_rows = self._filter_display_fields(
                columns, rows_list
            )
            if filtered_columns:
                columns = filtered_columns
                rows_list = filtered_rows

            # 根据配置决定平台列显示中文还是英文
            rows_list = self._apply_platform_name_toggle(columns, rows_list, platform_configs)

            return {
                "query_result": {
                    "columns": columns,
                    "rows": rows_list,
                }
            }

        except Exception as exc:
            get_logger().error("收集CPS数据质量监控数据失败: %s", str(exc), exc_info=True)
            raise

    def _filter_display_fields(
        self, columns: List[str], rows: List[List[Any]]
    ) -> Tuple[Optional[List[str]], Optional[List[List[Any]]]]:
        """
        根据 display_fields 过滤列，将列标题映射到SQL输出的中文列
        """
        if not isinstance(self.params, dict):
            return None, None
        display_fields = self.params.get("display_fields")
        if not display_fields:
            return None, None

        field_labels = self._get_field_labels()
        column_index_map = {col: idx for idx, col in enumerate(columns)}

        get_logger().debug(
            "[CPS数据质量监控] _filter_display_fields - columns: %s, display_fields: %s",
            columns,
            display_fields,
        )

        filtered_columns: List[str] = []
        selected_indexes: List[int] = []
        for field in display_fields:
            label = field_labels.get(field)
            if not label:
                continue
            column_idx = column_index_map.get(label)
            if column_idx is None:
                continue
            filtered_columns.append(label)
            selected_indexes.append(column_idx)

        if not filtered_columns:
            return None, None

        filtered_rows: List[List[Any]] = []
        for row in rows:
            filtered_row = []
            for idx in selected_indexes:
                filtered_row.append(row[idx] if idx < len(row) else None)
            filtered_rows.append(filtered_row)
        
        return filtered_columns, filtered_rows

    def _apply_platform_name_toggle(
        self,
        columns: List[str],
        rows: List[List[Any]],
        platform_configs: Dict[str, Any],
    ) -> List[List[Any]]:
        """根据 use_chinese_platform_name 开关，将平台代码映射为中文名称"""
        if not isinstance(self.params, dict):
            return rows

        use_chinese = self.params.get("use_chinese_platform_name")
        # 默认开启中文；只有显式关闭时才不用
        if isinstance(use_chinese, str):
            use_chinese = use_chinese.lower() in ("1", "true", "yes", "y", "on")
        elif isinstance(use_chinese, (int, float)):
            use_chinese = bool(use_chinese)
        if use_chinese is None:
            use_chinese = True
        if not use_chinese:
            return rows

        if not isinstance(platform_configs, dict) or not platform_configs:
            return rows

        # 构建 平台代码 -> 中文名 映射（用于“平台”列）
        platform_name_map: Dict[str, str] = {}
        # 构建 汇总平台代码 -> 中文名 映射（用于“平台汇总”列）
        summary_name_map: Dict[str, str] = {}

        for code, cfg in platform_configs.items():
            if not isinstance(cfg, dict):
                continue

            name = cfg.get("name")
            if isinstance(name, str) and name.strip():
                platform_name_map[code] = name.strip()

            # 汇总平台名称：使用 platform_all 作为 key，summary_name 优先，其次 name
            agg_code = cfg.get("platform_all") or code
            if not isinstance(agg_code, str):
                continue
            summary_name = cfg.get("summary_name") or name
            if isinstance(summary_name, str) and summary_name.strip():
                # 只在第一次设置时生效，避免被后续相同 platform_all 覆盖
                summary_name_map.setdefault(agg_code, summary_name.strip())

        if not platform_name_map and not summary_name_map:
            return rows

        # 列标题（中文）是通过 field_labels 控制的
        field_labels = self._get_field_labels()
        platform_col_name = field_labels.get("platform", "平台")
        platform_summary_col_name = field_labels.get("platform_summary", "平台汇总")

        try:
            platform_idx = columns.index(platform_col_name)
        except ValueError:
            platform_idx = None

        try:
            platform_summary_idx = columns.index(platform_summary_col_name)
        except ValueError:
            platform_summary_idx = None

        if platform_idx is None and platform_summary_idx is None:
            return rows

        new_rows: List[List[Any]] = []
        for row in rows:
            new_row = list(row)

            # 平台列：dy/dyc -> 抖音/抖音招商 等
            if platform_idx is not None and platform_idx < len(new_row):
                code = new_row[platform_idx]
                if isinstance(code, str):
                    if code in platform_name_map:
                        new_row[platform_idx] = platform_name_map[code]
                    elif code.upper() == "ALL":
                        # 汇总行的平台列显示为“平台”（或自定义字段标签）
                        new_row[platform_idx] = platform_col_name

            # 平台汇总列：使用汇总平台中文名（tb->淘宝, dy->抖音 等；ALL 保留）
            if platform_summary_idx is not None and platform_summary_idx < len(new_row):
                val = new_row[platform_summary_idx]
                if isinstance(val, str) and val in summary_name_map:
                    new_row[platform_summary_idx] = summary_name_map[val]

            new_rows.append(new_row)

        return new_rows

    def validate_params(self) -> List[str]:
        errors = []

        if "data_source_config" not in self.params:
            errors.append("缺少数据源配置")
        else:
            data_source_config = self.params["data_source_config"]
            if not data_source_config.get("data_source_id"):
                errors.append("数据源ID不能为空")

        platform_selection = self.params.get("platform_selection") or self.params.get(
            "platform_filter"
        )
        if not platform_selection:
            errors.append("缺少平台筛选配置")
        else:
            selected_platforms = platform_selection.get("selected_platforms", [])
            if not selected_platforms:
                errors.append("至少需要选择一个平台")
            else:
                # 从数据库读取平台配置进行验证
                try:
                    platform_configs = self._get_platform_configs()
                    invalid_platforms = [
                        p for p in selected_platforms if p not in platform_configs
                    ]
                    if invalid_platforms:
                        errors.append(f"无效的平台: {', '.join(invalid_platforms)}")
                except Exception as e:
                    get_logger().warning(f"验证平台配置时出错: {e}")
                    errors.append(f"平台配置验证失败: {str(e)}")

        if "time_filter" not in self.params:
            errors.append("缺少时间筛选配置")
        else:
            time_filter = self.params["time_filter"]
            time_range_type = time_filter.get("time_range_type")
            if not time_range_type:
                errors.append("请选择时间范围类型")
            elif time_range_type == "custom":
                if not time_filter.get("time_start") or not time_filter.get("time_end"):
                    errors.append("自定义时间范围需要设置开始和结束时间")

        if "threshold_config" not in self.params:
            errors.append("缺少阈值配置")
        else:
            threshold_config = self.params["threshold_config"]
            sync_rate_threshold = threshold_config.get("sync_rate_threshold")
            if (
                sync_rate_threshold is None
                or sync_rate_threshold < 0
                or sync_rate_threshold > 100
            ):
                errors.append("同步率阈值必须在0-100之间")

        return errors

    def get_platform_list(self) -> List[Dict[str, str]]:
        """从数据库配置获取平台列表"""
        try:
            platform_configs = self._get_platform_configs()
            return [
                {"value": key, "label": value.get("name", key)}
                for key, value in platform_configs.items()
            ]
        except Exception as e:
            get_logger().error(f"获取平台列表失败: {e}", exc_info=True)
            return []

    def get_supported_time_range_types(self) -> List[str]:
        return ["all", "yesterday", "recent", "last_month", "custom"]

    def generate_summary(self, data: Dict[str, Any]) -> Dict[str, Any]:
        try:
            query_result = data.get("query_result", {})
            columns = query_result.get("columns", [])
            rows = query_result.get("rows", [])

            if not rows:
                return {
                    "total_records": 0,
                    "platform_count": 0,
                    "avg_sync_rate": 0,
                    "status": "无数据",
                }

            total_records = len(rows)
            platform_count = len({row[1] for row in rows if len(row) > 1})

            sync_rates = []
            for row in rows:
                if len(row) > 3 and row[3] and isinstance(row[3], (int, float, str)):
                    try:
                        sync_rate = float(str(row[3]).replace("%", ""))
                        sync_rates.append(sync_rate)
                    except (ValueError, TypeError):
                        continue

            avg_sync_rate = sum(sync_rates) / len(sync_rates) if sync_rates else 0

            return {
                "total_records": total_records,
                "platform_count": platform_count,
                "avg_sync_rate": round(avg_sync_rate, 2),
                "status": "正常",
            }

        except Exception as exc:
            get_logger().error("生成摘要失败: %s", str(exc), exc_info=True)
            return {
                "total_records": 0,
                "platform_count": 0,
                "avg_sync_rate": 0,
                "status": "错误",
            }

    def preview_sql(self) -> str:
        """生成预览 SQL"""
        try:
            data_source_config = self.params.get("data_source_config", {})
            data_source_id = data_source_config.get("data_source_id")

            if not data_source_id:
                raise ValueError("数据源ID不能为空")

            platform_selection = self.params.get("platform_selection") or self.params.get(
                "platform_filter", {}
            )
            selected_platforms = platform_selection.get("selected_platforms", ["dy"])
            include_summary = platform_selection.get("include_summary", True)

            time_filter = self.params.get("time_filter", {})
            threshold_config = self.params.get("threshold_config", {})
            sync_rate_threshold = threshold_config.get("sync_rate_threshold", 80)

            # 构建时间条件
            time_condition = self._build_time_condition(time_filter)

            # 从数据库读取平台配置
            platform_configs = self._get_platform_configs()

            # 构建源表 UNION SQL（按用户选择的具体 platform 统计）
            source_union_sql = self._build_source_union_sql(
                selected_platforms, platform_configs, time_condition
            )

            if not source_union_sql:
                return (
                    "SELECT '无数据' AS '平台', '无平台选择' AS '平台汇总', '无数据' AS '原始层(源表)', "
                    "0 AS '原始层商品总数', 0 AS '目标表商品总数', 0 AS '目标表上架商品数', "
                    "0 AS '目标表下架商品数', '0%' AS '同步率', '无数据' AS '数据状态', 1 AS sort_order"
                )

            # 获取字段标签
            field_labels = self._get_field_labels()

            # 构建平台汇总 SQL
            platform_summary_sql = ""
            if include_summary:
                platform_summary_sql = f"""
SELECT
    'ALL' AS `{field_labels.get('platform', '平台')}`,
    '全平台汇总' AS `{field_labels.get('platform_summary', '平台汇总')}`,
    '所有平台汇总' AS `{field_labels.get('source_layer', '原始层(源表)')}`,
    SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) AS `{field_labels.get('source_total_count', '原始层商品总数')}`,
    SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) AS `{field_labels.get('dws_total_count', '目标表商品总数')}`,
    SUM(`{field_labels.get('dws_on_sale_count', '目标表上架商品数')}`) AS `{field_labels.get('dws_on_sale_count', '目标表上架商品数')}`,
    SUM(`{field_labels.get('dws_off_sale_count', '目标表下架商品数')}`) AS `{field_labels.get('dws_off_sale_count', '目标表下架商品数')}`,
    CONCAT(
        ROUND(
            CASE
                WHEN SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) > 0
                THEN (SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) * 100.0 / SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`))
                ELSE 0
            END,
            2
        ),
        '%'
    ) AS `{field_labels.get('sync_rate', '同步率')}`,
    CASE
        WHEN SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`) = 0 THEN '无原始数据'
        WHEN SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) = 0 THEN '未同步'
        WHEN (SUM(`{field_labels.get('dws_total_count', '目标表商品总数')}`) * 100.0 / SUM(`{field_labels.get('source_total_count', '原始层商品总数')}`)) < {sync_rate_threshold} THEN '同步率过低'
        ELSE '正常'
    END AS `{field_labels.get('data_status', '数据状态')}`,
    0 AS sort_order
FROM platform_detail

UNION ALL

"""

            # 获取 SQL 模板
            sql_template = self._get_sql_template()

            # 构建模板上下文
            # DWS 表查询使用用户选择的具体 platform，platform_all 仅用于展示
            formatted_platforms = ", ".join([f"'{p}'" for p in selected_platforms])
            context = {
                "source_union_sql": source_union_sql,
                "selected_platforms": formatted_platforms,  # DWS 表使用具体 platform
                "time_condition": time_condition,
                "sync_rate_threshold": sync_rate_threshold,
                "platform_summary_sql": platform_summary_sql,
                "platform_label": field_labels.get("platform", "平台"),
                "platform_summary_label": field_labels.get("platform_summary", "平台汇总"),
                "source_layer_label": field_labels.get("source_layer", "原始层(源表)"),
                "source_total_label": field_labels.get("source_total_count", "原始层商品总数"),
                "dws_total_label": field_labels.get("dws_total_count", "目标表商品总数"),
                "dws_on_sale_label": field_labels.get("dws_on_sale_count", "目标表上架商品数"),
                "dws_off_sale_label": field_labels.get("dws_off_sale_count", "目标表下架商品数"),
                "sync_rate_label": field_labels.get("sync_rate", "同步率"),
                "data_status_label": field_labels.get("data_status", "数据状态"),
            }

            # 渲染 SQL 模板
            final_sql = self._render_sql_template(sql_template, context)
            return final_sql

        except Exception as exc:
            get_logger().error("生成预览SQL失败: %s", str(exc), exc_info=True)
            raise
